package a.ft;

import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.connector.datagen.table.DataGenConnectorOptions;
import org.apache.flink.table.api.*;

public class MyTableApiAndSQLTest {
    public static void main(String[] args) {
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment environment = TableEnvironment.create(settings);
        environment.getConfig().getConfiguration().setString("parallelism.default", "1");
//        读取数据源
//
        environment.createTemporaryTable("SourceTable", TableDescriptor.forConnector("datagen")
                .schema(Schema.newBuilder()
                        .column("f0", DataTypes.STRING())
                        .build())
                .option(DataGenConnectorOptions.ROWS_PER_SECOND, 10l)


                .build());
        environment.executeSql(" create table SinkTable(f0 string)  with ('connector' = 'print')");
        Table table = environment.from("SourceTable");
//        table.execute().print();
//        environment.sqlQuery(" select * from  SourceTable").execute().print();

        table.executeInsert("SinkTable");

    }
}
